package com.zhang.spark_streaming.kafka

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.util.Properties
import scala.collection.mutable.ListBuffer
import scala.util.Random

/**
 * @title:
 * @author: zhang
 * @date: 2021/12/12 23:16 
 */
object SparkStreaming_Kafka_p {

  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming")
    val ssc = new StreamingContext(sparkConf,Seconds(3))
    // 创建配置对象
    val prop = new Properties()
    // 添加配置
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092,hadoop104:9092")
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

    // 根据配置创建Kafka生产者
    val producer = new KafkaProducer[String, String](prop)

    while (true){

      mockData().foreach(
        data=>{
          val value = new ProducerRecord[String, String]("spark", data)
          producer.send(value)
        }
      )
      Thread.sleep(2000)
    }

    ssc.start()
    ssc.awaitTermination()
  }

  def mockData() ={
    val list = ListBuffer[String]()
    val areaList = List("华北","华东","华南")
    val cityList = List("北京","上海","深圳")

    for(i <- 1 to 30){
      val area = areaList(new Random().nextInt(3))
      val city = cityList(new Random().nextInt(3))
      val userid = new Random().nextInt(6)+1
      val adid = new Random().nextInt(6)+1


      list.append(s"${System.currentTimeMillis()} $area $city $userid $adid")
    }
    list
  }

}
